package com.forezp.configclient.rabbitConfig;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ChannelListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.ErrorHandler;

import java.io.IOException;
import java.util.UUID;

/**
 * @version V1.0  </br>
 * @ClassName MqRabbitConfig </br>
 * @Description TODO  </br>
 * @Author tonywang</br>
 * @DATE 11/21/2018 5:17 PM</br>
 */
@Configuration
public class MqRabbitConfig {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Value("${spring.rabbitmq.addresses}")
    private String addresses;
    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirms;

    @Autowired
    @Bean
    public ConnectionFactory amqpConnectionFactory(ConnectionListener connectionListener,
                                                   RecoveryListener recoveryListener,
                                                   ChannelListener channelListener) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        //设置了rabbitmq的地址、端口，集群部署的情况下可填写多个，“,”分隔。
//        connectionFactory.setAddresses(addresses);
        connectionFactory.setHost(host);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        //设置缓存模式，共有两种，CHANNEL和CONNECTION模式。
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        //设置每个Connection中（注意是每个Connection）可以缓存的Channel数量，注意只是缓存的Channel数量，不是Channel的数量上限，操作rabbitmq之前（send/receive message等）要先获取到一个Channel，获取Channel时会先从缓存中找闲置的Channel，如果没有则创建新的Channel，当Channel数量大于缓存数量时，多出来没法放进缓存的会被关闭。
        connectionFactory.setChannelCacheSize(25);
        //当这个值大于0时，channelCacheSize不仅是缓存数量，同时也会变成数量上限，从缓存获取不到可用的Channel时，不会创建新的Channel，会等待这个值设置的毫秒数，到时间仍然获取不到可用的Channel会抛出AmqpTimeoutException异常。
        connectionFactory.setChannelCheckoutTimeout(0);
        //producer端的消息确认机制（confirm和return），设为true后开启相应的机制
        //ReturnCallback - 只有在消息进入exchange但没有进入queue时才会调用。
        connectionFactory.setPublisherReturns(true);
        //ConfirmCallback - 每一条发出的消息都会调用ConfirmCallback；
        connectionFactory.setPublisherConfirms(publisherConfirms);

        connectionFactory.addConnectionListener(connectionListener);

        connectionFactory.addChannelListener(channelListener);

        connectionFactory.setRecoveryListener(recoveryListener);
        //仅在CONNECTION模式使用，设置Connection的缓存数量。
        //connectionFactory.setConnectionCacheSize(1);
        //仅在CONNECTION模式使用，设置Connection的数量上限
        //connectionFactory.setConnectionLimit(Integer.MAX_VALUE);

        return connectionFactory;
    }
    //consumer端如果通过@RabbitListener注解的方式接收消息，不需要这个bean
    @Autowired
    @Bean
    public AmqpTemplate amqpTemplate(ConnectionFactory amqpConnectionFactory,
                                     RabbitTemplate.ReturnCallback returnCallback,
                                     RabbitTemplate.ConfirmCallback confirmCallback,
                                     RetryTemplate retryTemplate,
                                     MessageConverter messageConverter){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        //设置spring-amqp的ConnectionFactory
        rabbitTemplate.setConnectionFactory(amqpConnectionFactory);
        //设置重试机制
        rabbitTemplate.setRetryTemplate(retryTemplate);
        //设置MessageConverter，用于java对象与Message对象（实际发送和接收的消息对象）之间的相互转换
        rabbitTemplate.setMessageConverter(messageConverter);
        rabbitTemplate.setReturnCallback(returnCallback);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //设为true使ReturnCallback生效。
        rabbitTemplate.setMandatory(false);
        //打开或关闭Channel的事务
        rabbitTemplate.setChannelTransacted(false);
        //RabbitTemplate在创建Connection时，会根据这个boolean参数选择使用ConnectionFactory本身或者ConnectionFactory中的publisherConnectionFactory（也是一个ConnectionFactory）来创建
        rabbitTemplate.setUsePublisherConnection(true);
        return rabbitTemplate;
    }

    @Bean
    public ErrorHandler errorHandler(){
        return new ConditionalRejectingErrorHandler();
    }

    //这个bean仅在consumer端通过@RabbitListener注解的方式接收消息时使用，每一个@RabbitListener注解的方法都会由这个RabbitListenerContainerFactory创建一个MessageListenerContainer，负责接收消息。
    @Autowired
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(CachingConnectionFactory cachingConnectionFactory,
                                                                               ErrorHandler errorHandler,
                                                                               MessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setErrorHandler(errorHandler);
        factory.setMessageConverter(messageConverter);
        //设置consumer端的应答模式，共有三种：NONE、AUTO、MANUAL。
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        //设置每个MessageListenerContainer将会创建的Consumer的最小数量，默认是1个
        factory.setConcurrentConsumers(1);
        //设置每个MessageListenerContainer将会创建的Consumer的最大数量，默认等于最小数量
        factory.setMaxConcurrentConsumers(1);
        //设置每次请求发送给每个Consumer的消息数量。
        factory.setPrefetchCount(250);
        //打开或关闭Channel的事务
        factory.setChannelTransacted(false);
        //设置事务当中可以处理的消息数量
        factory.setTxSize(1);
        //设置当rabbitmq收到nack/reject确认信息时的处理方式，设为true，扔回queue头部，设为false，丢弃。
        factory.setDefaultRequeueRejected(true);
        //实现ErrorHandler接口设置进去，所有未catch的异常都会由ErrorHandler处理。
        factory.setErrorHandler(errorHandler);
        return factory;
    }


    @Bean
    public MessageConverter messageConverter(){
//        return new Jackson2JsonMessageConverter();
        return new SimpleMessageConverter();
    }

    public void controlChannel(CachingConnectionFactory amqpConnectionFactory){
        Channel channel = amqpConnectionFactory.createConnection().createChannel(true);
        try {
            //channel.txSelect();上面createChannel已经设为true了，这句可以去掉
            Object event = new String("566");
            channel.basicPublish("xxx", "xxx", new AMQP.BasicProperties(), JSON.toJSONString(event).getBytes());
            channel.txCommit();
        } catch (IOException e) {
            try {
                channel.txRollback();
            } catch (IOException e1) {
            }
        } finally {
            try {
                channel.close();
            } catch (Exception e) {
            }
        }
    }
    //ReturnCallback - 只有在消息进入exchange但没有进入queue时才会调用。
    @Bean
    public RabbitTemplate.ReturnCallback returnCallback() {
        return new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                logger.info("return call back");
            }
        };
    }
    //ConfirmCallback - 每一条发出的消息都会调用ConfirmCallback；
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback() {
        return new RabbitTemplate.ConfirmCallback() {
            @Override
            //correlationData - RabbitTemplate的send系列方法中有带这个参数的，如果传了这个参数，会在回调时拿到；
            //ack - 消息进入exchange，为true，未能进入exchange，为false，由于Connection中断发出的消息进入exchange但没有收到confirm信息的情况，也会是false；
            //cause - 消息发送失败时的失败原因信息
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                logger.info("confirm call back");
            }
        };
    }

    //ChannelListener接口，监听Channel的创建和异常关闭。
    @Bean
    public ChannelListener channelListener() {
        return new ChannelListener() {
            @Override
            public void onCreate(Channel channel, boolean transactional) {
                logger.info("channel number:{}, nextPublishSqlNo:{}",
                        channel.getChannelNumber(),
                        channel.getNextPublishSeqNo());
            }
            @Override
            public void onShutDown(ShutdownSignalException signal) {
                logger.error("channel shutdown, reason:{}, errorLevel:{}",
                        signal.getReason().protocolMethodName(),
                        signal.isHardError() ? "connection" : "channel");
            }
        };
    }

    //BlockedListener监听Connection的block和unblock。
    @Bean
    public BlockedListener blockedListener() {
        return new BlockedListener() {
            @Override
            public void handleBlocked(String reason) throws IOException {
                logger.info("connection blocked, reason:{}", reason);
            }
            @Override
            public void handleUnblocked() throws IOException {
                logger.info("connection unblocked");
            }
        };
    }

    //ConnectionListener监听Connection的创建、关闭和异常终止。
    @Bean
    public ConnectionListener connectionListener() {
        return new ConnectionListener() {
            @Override
            public void onCreate(org.springframework.amqp.rabbit.connection.Connection connection) {
                logger.info("connection created.");
            }

            @Override
            public void onClose(org.springframework.amqp.rabbit.connection.Connection connection) {
                logger.info("connection closed.");
            }

            public void onShutDown(ShutdownSignalException signal) {
                logger.error("connection shutdown, reason:{}, errorLevel:{}",
                        signal.getReason().protocolMethodName(),
                        signal.isHardError() ? "connection" : "channel");
            }
        };
    }

    //RecoveryListener监听开始自动恢复Connection、自动恢复连接完成。
    @Bean
    public RecoveryListener recoveryListener() {
        return new RecoveryListener() {
            @Override
            public void handleRecovery(Recoverable recoverable) {
                logger.info("automatic recovery completed");
            }
            @Override
            public void handleRecoveryStarted(Recoverable recoverable) {
                logger.info("automatic recovery started");
            }
        };
    }

    //重试机制主要是解决网络不稳导致连接中断的问题。所以其实并不是重新发送消息，而是重新建立。
    //RetryTemplate与spring-amqp及rabbitmq都没有关系，这是spring-retry中的类。以上示例中使用了最简单的重试策略，不断重试，直到Integer.MAX_VALUE次为止。
    @Bean
    public RetryTemplate retryTemplate(){
        RetryTemplate retryTemplate = new RetryTemplate();
        SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(Integer.MAX_VALUE);
        retryTemplate.setRetryPolicy(simpleRetryPolicy);
        return retryTemplate;
    }
}
